package com.jianying.day04;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * ClassName: Flink05_Union
 * Package: com.atguigu.day04
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/2 13:59
 * @Version: 1.0
 */
public class Flink06_Connect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> socketDS1 = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> socketDS2 = env.socketTextStream("localhost", 8889);

        SingleOutputStreamOperator<Integer> intDS2 = socketDS2.map(Integer::valueOf);

        // TODO Connect 合流
        ConnectedStreams<String, Integer> connectDS = socketDS1.connect(intDS2);

        // TODO 打印前处理一下
        SingleOutputStreamOperator<String> mapDS = connectDS.map(
                new CoMapFunction<String, Integer, String>() {
                    @Override
                    public String map1(String value) throws Exception {
                        return "字符串：" + value;
                    }

                    @Override
                    public String map2(Integer value) throws Exception {
                        return "数字：" + value;
                    }
                }
        );

        mapDS.print();


        env.execute();
    }
}
